package com.ds.infrastructure.hyperspace.container.endpoint.connector;

import com.ds.infrastructure.hyperspace.common.TargetEndpointStream;
import com.ds.infrastructure.hyperspace.common.TargetEndpointStreamType;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.NettyTargetEndpointChannelInitializer;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoop;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.*;

import static com.ds.infrastructure.hyperspace.common.TargetEndpointStreamStatus.CONNECT_IGNORE;
import static com.ds.infrastructure.hyperspace.common.TargetEndpointStreamType.POOLED;

/**
 * Note: this class is not allowed switch thread, you must run it on its own thread.
 *
 * @author  Weishaoying
 */
@Slf4j
public class NettyEndpointNodeStreamManager {

    public static final TargetEndpointStream NOT_ALLOW_CREATE_STREAM =
            new EmptyTargetEndpointStream(CONNECT_IGNORE);

    @Getter
    private final NettyContainer nettyContainer;

    @Getter
    private final TargetEndpointNode targetEndpointNode;

    @Getter
    private final Bootstrap bootstrap;

    @Getter
    private final HyperspaceConnector hyperspaceConnector;

    private final HyperspaceStreamTimer hyperspaceStreamTimer;

    private final List<NettyEndpointNodeStream> streamPool;

    private final Map<String, NettyEndpointNodeStream> allStreamMap;

    private int acquireIndex = 0;

    public NettyEndpointNodeStreamManager(NettyContainer nettyContainer,
                                          EventLoop entryEventLoop, TargetEndpointNode targetEndpointNode,
                                          HyperspaceConnector hyperspaceConnector,
                                          HyperspaceStreamTimer hyperspaceStreamTimer) {
        this.nettyContainer = nettyContainer;
        this.targetEndpointNode = targetEndpointNode;
        this.hyperspaceConnector = hyperspaceConnector;
        this.hyperspaceStreamTimer = hyperspaceStreamTimer;
        this.bootstrap = new Bootstrap().group(entryEventLoop)
                .channel(nettyContainer.getNettyTransportFactoryDelegate().socketChannelClass())
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
                        targetEndpointNode.getNodeConfig().getProtocolConfig().getConnectTimeout())
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new NettyTargetEndpointChannelInitializer(nettyContainer,
                        hyperspaceConnector, targetEndpointNode));
        this.streamPool = new ArrayList<>();
        this.allStreamMap = new HashMap<>();
    }

    public void initNettyTargetEndpointStreamPool(int poolSize) {
        long startTime = System.currentTimeMillis();
        int initSize = 0;
        while (initSize < poolSize) {
            TargetEndpointStream targetEndpointStream = createTargetEndpointStream(POOLED, false);
            if (CONNECT_IGNORE == targetEndpointStream.getStatus()) {
                break;
            }
            initSize++;
        }
        log.info("Init target node({}) {} connections in {} ms.", targetEndpointNode.getKey(),
                initSize, System.currentTimeMillis() - startTime);
    }

    public TargetEndpointStream acquireTargetEndpointStream() {
        if (targetEndpointNode.isAvailable() && !CollectionUtils.isEmpty(streamPool)) {
            if (acquireIndex >= streamPool.size()) {
                acquireIndex = 0;
            }
            NettyEndpointNodeStream tryAcquireStream = streamPool.get(acquireIndex++);
            if (tryAcquireStream.acquireWriteOperation()) {
                if (tryAcquireStream.isAvailableTargetNode()) {
                    return tryAcquireStream;
                }
                tryAcquireStream.closeTargetChannel();
            }
            int tempIndex = acquireIndex;
            int tempSize = streamPool.size();
            while (tempIndex < tempSize) {
                NettyEndpointNodeStream targetEndpointStream = streamPool.get(tempIndex++);
                if (targetEndpointStream.acquireWriteOperation()) {
                    if (targetEndpointStream.isAvailableTargetNode()) {
                        return targetEndpointStream;
                    }
                    targetEndpointStream.closeTargetChannel();
                }
                if (tempIndex == acquireIndex) {
                    break;
                } else if (tempIndex == tempSize) {
                    tempIndex = 0;
                    tempSize = acquireIndex;
                }
            }
        }
        return createTargetEndpointStream(POOLED, true);
    }

    public TargetEndpointStream createTargetEndpointStream(
            TargetEndpointStreamType streamType, boolean preAcquire) {
        if (!nettyContainer.allowIncrementTargetConnection(targetEndpointNode)) {
            return NOT_ALLOW_CREATE_STREAM;
        }
        NettyEndpointNodeStream nettyEndpointNodeStream = new NettyEndpointNodeStream(
                preAcquire, this);
        nettyEndpointNodeStream.setStreamType(streamType);
        return nettyEndpointNodeStream;
    }

    public void onStreamConnectSuccess(NettyEndpointNodeStream stream) {
        hyperspaceStreamTimer.addEvent(stream);
        allStreamMap.putIfAbsent(stream.getEventId(), stream);
        if (POOLED.equals(stream.getStreamType())) {
            streamPool.add(stream);
        }
        if (!targetEndpointNode.isAvailable()) {
            recoverClosedStreamPool();
            targetEndpointNode.incrementPassCount(stream.getRemoteAddress());
        }
    }

    public void onStreamConnectFailed(NettyEndpointNodeStream stream) {
        allStreamMap.remove(stream.getEventId());
        if (targetEndpointNode.isAvailable()) {
            recoverClosedStreamPool();
            targetEndpointNode.incrementFailCount(stream.getRemoteAddress());
        }
    }

    public void onStreamRelease(NettyEndpointNodeStream stream) {
        hyperspaceConnector.onStreamRelease();
    }

    public void onStreamClose(NettyEndpointNodeStream stream) {
        hyperspaceStreamTimer.removeEvent(stream);
        allStreamMap.remove(stream.getEventId());
        if (nettyContainer.needHoldTargetConnection(targetEndpointNode) &&
                nettyContainer.allowIncrementTargetConnection(targetEndpointNode)) {
            stream.connect();
        }
        recoverClosedStreamPool();
    }

    void recoverClosedStreamPool() {
        if (streamPool.isEmpty()) {
            return;
        }
        Iterator<NettyEndpointNodeStream> streamIterator = streamPool.iterator();
        while (streamIterator.hasNext()) {
            NettyEndpointNodeStream tempStream = streamIterator.next();
            if (tempStream.isChannelActive()) {
                continue;
            }
            streamIterator.remove();
        }
    }

    void shutdownNodeManager() {
        if (!allStreamMap.isEmpty()) {
            for (Map.Entry<String, NettyEndpointNodeStream> streamEntry : allStreamMap.entrySet()) {
                NettyEndpointNodeStream targetEndpointNodeStream = streamEntry.getValue();
                targetEndpointNodeStream.closeTargetChannel();
            }
        }
        if (!streamPool.isEmpty()) {
            streamPool.clear();
        }
    }

}
